package demo.spark.local;

import demo.utils.JsonUtils;
import demo.utils.LogClassifyUtil;
import demo.vo.LogStatModel;
import demo.vo.PlayStatResultModel;
import demo.vo.XmlyLogSpan;
import kafka.serializer.StringDecoder;
import org.apache.commons.lang.StringUtils;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.api.java.function.VoidFunction;
import org.apache.spark.streaming.Duration;
import org.apache.spark.streaming.api.java.JavaDStream;
import org.apache.spark.streaming.api.java.JavaPairDStream;
import org.apache.spark.streaming.api.java.JavaPairInputDStream;
import org.apache.spark.streaming.api.java.JavaStreamingContext;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.Serializable;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;

import org.apache.spark.streaming.kafka.KafkaUtils;
import scala.Tuple2;
import demo.vo.*;
import demo.utils.*;

public class LogStatStreaming {

    private static final Logger logger = LoggerFactory.getLogger(LogStatStreaming.class);

    public static void main(String[] args) throws InterruptedException {
        logger.info("streaming process start !!!");
        SparkConf sparkConf = new SparkConf().setAppName("xdcs-log-stat");
        Map<String, String> sparkConfProperties = getSparkConfig();
        for (Map.Entry<String, String> entry : sparkConfProperties.entrySet()) {
            sparkConf.set(entry.getKey(), entry.getValue());
        }
        JavaStreamingContext javaStreamingContext = new JavaStreamingContext(sparkConf, new Duration(60000));//spark batch handler time;
        JavaPairDStream<String, String> inputWindowData = getInputJavaPairDStream(javaStreamingContext);
        doProcess(inputWindowData);
        javaStreamingContext.start();
        javaStreamingContext.awaitTermination();
    }

    public static Map<String, String> getSparkConfig() {
        return new HashMap<>();
    }

    public static String get(String key) {
        return getSparkConfig().get(key);
    }

    public static JavaPairDStream<String, String> getInputJavaPairDStream(JavaStreamingContext jsc) {
        String groupId = "xdcs-spark-log-stat-0904";
        Set<String> topics = new HashSet<String>();
        topics.add("log");
        String brokerList = get("kafka.metadata.broker.list");
        String rest = get("kafka.auto.offset.reset");
        int windowDuration = 60000;
        int slideDuration = 60000;
        Map<String, String> kafkaParams = new HashMap<String, String>();
        kafkaParams.put("metadata.broker.list", brokerList);
        kafkaParams.put("group.id", groupId);
        kafkaParams.put("auto.offset.reset", rest);
        JavaPairInputDStream<String, String> messages = KafkaUtils.createDirectStream(jsc, String.class, String.class, StringDecoder.class, StringDecoder.class, kafkaParams, topics);
        return messages.window(new Duration(windowDuration), new Duration(slideDuration));//每60秒统计最近60秒的数据 ONE_MINUTES_BATCH
    }

    static void doProcess(JavaPairDStream<String, String> inputData) {
        int repartitionNum = Integer.parseInt(get("spark.repartion.num"));
        JavaPairDStream<String, String> repartitionDStream;
        if (repartitionNum > 0) {
            repartitionDStream = inputData.repartition(repartitionNum);
        } else {
            repartitionDStream = inputData;
        }
        JavaDStream<LogStatModel> nacDStream = repartitionDStream.flatMap(
                new FlatMapFunction<Tuple2<String, String>, LogStatModel>() {
                    @Override
                    public Iterator<LogStatModel> call(Tuple2<String, String> t) throws Exception {
                        List<LogStatModel> playStatModels = new ArrayList<LogStatModel>();
                        LogStatModel playStatModel = getModels(t);
                        if (playStatModel != null) {
                            playStatModels.add(playStatModel);
                        }

                        return playStatModels.iterator();
                    }
                });
        nacDStream.persist();
        processPlayDStream(nacDStream);

    }

    private static void processPlayDStream(JavaDStream<LogStatModel> appServerStatData) {
        JavaPairDStream<String, PlayStatResultModel> playPairDStream = appServerStatData
                .filter(new Function<LogStatModel, Boolean>() {
                    private static final long serialVersionUID = -1506745184525363167L;

                    @Override
                    public Boolean call(LogStatModel v1) throws Exception {
                        String key = v1.getEventType();
                        String[] infos = null;

                        if (StringUtils.isNotBlank(key)) {
                            infos = key.split("\\|\\|");
                        }
                        if (StringUtils.equals(v1.getApp(), "southgate") && infos != null && infos.length == 4) {
                            return false;
                        }

                        //info/warn/debug日志过滤掉
                        if (StringUtils.equals(v1.getLevel(), "INFO")
                                || StringUtils.equals(v1.getLevel(), "WARN")
                                || StringUtils.equals(v1.getLevel(), "DEBUG")) {
                            return false;
                        }

                        if (StringUtils.isNotBlank(v1.getApp())
                                && StringUtils.isNotBlank(v1.getHostName())) {
                            return true;
                        }
                        return false;
                    }
                }).mapToPair(new PairFunction<LogStatModel, String, PlayStatResultModel>() {
                    private static final long serialVersionUID = -1418100668369018875L;

                    @Override
                    public Tuple2<String, PlayStatResultModel> call(LogStatModel t) throws Exception {
                        String key = null;
                        PlayStatResultModel playStatResultModel = new PlayStatResultModel();
                        key = t.getReduceKey();
                        playStatResultModel.setPlayStatTypeEnum(PlayStatTypeEnum.LOG_STAT);
                        playStatResultModel.setCount(1);
                        playStatResultModel.setTimeStamp(t.getTimestamp());
                        playStatResultModel.setKey(key);
                        playStatResultModel.setStatTime(t.getTimestamp());
                        return new Tuple2<String, PlayStatResultModel>(key, playStatResultModel);
                    }
                }).reduceByKey(new Function2<PlayStatResultModel, PlayStatResultModel, PlayStatResultModel>() {
                    @Override
                    public PlayStatResultModel call(PlayStatResultModel v1, PlayStatResultModel v2) throws Exception {
                        v1.setCount(v1.getCount() + v2.getCount());
                        return v1;
                    }
                }, 64)
                .mapToPair(new PairFunction<Tuple2<String, PlayStatResultModel>, String, PlayStatResultModel>() {
                    @Override
                    public Tuple2<String, PlayStatResultModel> call(Tuple2<String, PlayStatResultModel> t) throws Exception {
                        if (t._2.getPlayStatTypeEnum().equals(PlayStatTypeEnum.LOG_STAT)) {
                            XdcsSparkUtils.dealWithLogResult(t._1, t._2);
                        } else {
                            XdcsSparkUtils.dealWithResult(t._1, t._2);
                        }

                        StatDataPoint statDataPoint = new StatDataPoint();
                        statDataPoint.setValue(String.valueOf(t._2.getCount()));
                        return new Tuple2<String, PlayStatResultModel>(t._1, t._2);
                    }
                });

        playPairDStream.foreachRDD(new VoidFunction<JavaPairRDD<String, PlayStatResultModel>>() {
            @Override
            public void call(JavaPairRDD<String, PlayStatResultModel> v1) throws Exception {
                List<Tuple2<String, PlayStatResultModel>> outputs = v1.collect();
                return;
            }
        });
    }

    private static LogStatModel getModels(Tuple2<String, String> tuple2) {
        String json = tuple2._2();
        LogStatModel playStatModel = null;
        try {
            XmlyLogSpan xmlySpan = JsonUtils.fromJson(json, XmlyLogSpan.class);
            if (StringUtils.contains(xmlySpan.getApp(), "rerank")) {
                //如果是rerank 开头的的logDetail不需要防止影响整体报警,他们错误日志的错误都相当大
                xmlySpan.setLogDetail("");
            }
            playStatModel = convertPlayModeFromSpan(xmlySpan);
        } catch (Throwable e) {
            logger.error(String.format("error while parse json, here the json is %s", json), e);
        }
        return playStatModel;
    }

    private static LogStatModel convertPlayModeFromSpan(XmlyLogSpan xmlySpan) {
        Long logTs = xmlySpan.getTimestamp();
        if (logTs == null || System.currentTimeMillis() - logTs >= 5 * 60 * 1000) {
            return null;
        }

        LogStatModel playStatModel = new LogStatModel();
        playStatModel.setApp(xmlySpan.getApp());
        playStatModel.setHostName(xmlySpan.getIp());
        playStatModel.setLevel(xmlySpan.getLevel());
        playStatModel.setLogIdentity(xmlySpan.getLogIdentity());
        playStatModel.setServiceId(xmlySpan.getServiceId());
        playStatModel.setTimestamp(System.currentTimeMillis());
        playStatModel.setEventType(xmlySpan.getEventType());

        String logIdentity = xmlySpan.getLogIdentity();
        String logDetail = xmlySpan.getLogDetail();

        String logType;
        if (StringUtils.equals(xmlySpan.getApp(), "nginx-log")) {
            logType = LogClassifyUtil.logNginxType(logIdentity);
        } else {
            logType = LogClassifyUtil.logType(logIdentity, logDetail);
        }

        playStatModel.setLogClassifyType(logType);
        return playStatModel;
    }


}
